/*
 * @(#)ReaderThread.java 1.0 2014/02/14
 *
 * Copyright (c) 2014 SinoSoft Refactored by filon51.
 * All rights reserved.
 *
 */
package com.sinosoft.cloud.dp.thread;


import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 读线程
 *
 * @author filon51
 * @since 2014-02-14
 */
public class ReaderThread extends Thread {
	private BusinessService m_businessService = null;
	private LinkedBlockingDeque m_readDataQueue = null;
	private int m_nProcessThreadCount = 0;
	private Logger m_log = Logger.getLogger(ReaderThread.class);

	public ReaderThread(BusinessService businessService,
						LinkedBlockingDeque readDataQueue, int processThreadCount) {
		m_businessService = businessService;
		m_readDataQueue = readDataQueue;
		m_nProcessThreadCount = processThreadCount;
	}

	public void run() {
		m_log.debug("ReaderThread start");
		m_businessService.prepareElement();
		HashMap hashElement = null;
		while (m_businessService.hasMoreElement()) {
			if(m_businessService.getExceptionFlag()) {//出现异常,就不需要再请求读数据了
				m_log.error("stop reading data,because it has one process dp occurs exception!");
				break;
			}
			hashElement = m_businessService.nextElement();
			try {
				m_readDataQueue.put(hashElement);
			} catch (Exception ex) {
				m_log.error("m_readDataQueue.put()", ex);
				ex.printStackTrace();
			}
		}

		// 放入监视对象，表示读数据已经结束
		for (int nIndex = 0; nIndex < m_nProcessThreadCount; nIndex++) {
			try {
				m_readDataQueue.put(MultiThreadExecutor.getOBJ_READ_GUARD());
			} catch (Exception ex) {
				m_log.error("m_readDataQueue.put() quiting ...", ex);
				ex.printStackTrace();
			}
		}
		m_log.debug("ReaderThread end");
	}
}
